package com.itc.bi.service.impl;

import cn.hutool.core.util.ObjectUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.IdWorker;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.itc.bi.constant.KafkaConstant;
import com.itc.bi.dto.DataSynDto;
import com.itc.bi.dto.DataSynPrarmDto;
import com.itc.bi.entity.*;
import com.itc.bi.handler.dataSet.Database;
import com.itc.bi.service.*;
import com.itc.bi.mapper.DataSynrecordMapper;
import com.itc.common.core.exception.ServiceException;
import com.itc.common.core.utils.StringUtils;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Lazy;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.CollectionUtils;
import org.springframework.util.ObjectUtils;

import java.util.ArrayList;
import java.util.Date;
import java.util.Iterator;
import java.util.List;

/**
 * @author songxiude
 * @description 针对表【tb_data_synrecord(数据同步创建记录表)】的数据库操作Service实现
 * @createDate 2024-07-04 15:32:32
 */
@Service
@Slf4j
public class DataSynrecordServiceImpl extends ServiceImpl<DataSynrecordMapper, DataSynrecordEntity>
        implements DataSynrecordService {


    @Autowired
    private ClickHouseJDBCService jdbcService;

    @Autowired
    private XxlJobInfoService xxlJobInfoService;

    @Autowired
    private TableService tableService;

    @Autowired
    private DirectDataSetService directDataSetService;

    @Autowired
    private DirectoryService directoryService;

    @Autowired
    private DataSetColumnService dataSetColumnService;

    @Autowired
    private Database database;

    @Autowired
    private DataSynApiService apiService;

    @Autowired
    XxlJobGroupService groupService;

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    @Value("${xxl.job.executor.appname}")
    private String jobName;

    /**
     * 数据同步任务生成
     *
     * @param prarmDto
     * @return
     */
    public List<String> generateDataSynTask(DataSynPrarmDto prarmDto) {

        //1、判断是否是数据集目录
        long count = directoryService.count(Wrappers.<DirectoryEntity>lambdaQuery()
                .eq(DirectoryEntity::getDtId, prarmDto.getDtId())
                .eq(DirectoryEntity::getIsDir, 0)
        );
        if (count == 0) {
            throw new ServiceException("数据集必须是目录");
        }
        //2、过滤处理过的数据集
        Iterator<DataSynDto> iterator = prarmDto.getTables().iterator();
        while (iterator.hasNext()) {
            DataSynDto table = iterator.next();
            if (prarmDto.getDataType().intValue() == 2 && StringUtils.isEmpty(table.getWhereColumn())) {
                throw new ServiceException("该数据表" + table.getTableName() + "为增量同步，增量条件字段不能为空！");
            }
            //int num = tableService.getDataSetNum(table.getTableId(), table.getTableName());
            long dscount = count(Wrappers.<DataSynrecordEntity>lambdaQuery().eq(DataSynrecordEntity::getTableId, table.getTableId()).eq(DataSynrecordEntity::getDirId, prarmDto.getDtId()));
            if (dscount > 0) {
                iterator.remove();
            }
        }
        //3、先创建回填表信息创建目标表任务调度生成
        //dataSynrecordService.addTableInfo(prarmDto);
        //修改新的
        List<String> tablenames = new ArrayList<>();
        for (DataSynDto table : prarmDto.getTables()) {
            String synTask = this.createDataSynTask(prarmDto.getDtId(), prarmDto.getDataType(), table);
            if(StringUtils.isNotEmpty(synTask)){
                tablenames.add(synTask);
            }else {
                tablenames.add(table.getTableName()+":失败");
            }


        }
        return tablenames;

    }


    /**
     * 保存业务数据
     *
     * @param prarmDto
     */
//    @Transactional(rollbackFor = Exception.class)
//    @Async("asyncServiceExecutor")
    public void addTableInfo(DataSynPrarmDto prarmDto) {
        //新的定时任务加入执行
        List<Integer> taskIds = new ArrayList<>();
        log.info("执行哪些表：{}", prarmDto.getTables().toArray());
        if (CollectionUtils.isEmpty(prarmDto.getTables())) {
            return;
        }
        XxlJobGroupEntity groupEntity = groupService.getOne(Wrappers.<XxlJobGroupEntity>lambdaQuery().eq(XxlJobGroupEntity::getAppName, jobName).last("limit 1"));
        prarmDto.getTables().forEach(table -> {
            //1、回填tb_direct_data_set
            DirectDataSetEntity directDataSet = new DirectDataSetEntity();
            //生成id回填
            long ddsid = IdWorker.getId();
            directDataSet.setId(ddsid);
            directDataSet.setDtId(prarmDto.getDtId());
            directDataSet.setTableName(table.getTableName());
            directDataSet.setType("DB");
            TableEntity tableEntity = tableService.getById(table.getTableId());
            directDataSet.setDsId(tableEntity.getDatasourceId());
            directDataSetService.save(directDataSet);
            //2、回填tb_directory
            DirectoryEntity directoryEntity = new DirectoryEntity();
            directoryEntity.setDtId(ddsid);
            directoryEntity.setPid(prarmDto.getDtId());
            directoryEntity.setTableName(table.getTargetTableName());
            directoryEntity.setIsDir(1);
            //设置名称先判断数据集是否存在
            long count = directoryService.count(Wrappers.<DirectoryEntity>lambdaQuery()
                    .eq(DirectoryEntity::getPid, prarmDto.getDtId())
                    .eq(DirectoryEntity::getName, table.getTableName())
            );
            //如果存在
            if (count > 0) {
                directoryEntity.setName(table.getTableName() + "1");
            } else {
                directoryEntity.setName(table.getTableName());
            }
            directoryService.save(directoryEntity);
            //3、回填tb_data_set_column
            List<DataSetColumnEntity> column = database.column(directDataSet);
            log.info("获取表字段信息：{}", column.toArray());
            dataSetColumnService.saveBatch(column);
            //4、先创建目标表
            //System.out.println(table.getTargetTableName());
            //jdbcService.createTableSourceDS(table.getTargetTableName(), table.getTabelId(), table.getWhereColumn());
            jdbcService.createTableSourceDS(table.getTargetTableName(), ddsid, table.getWhereColumn());

            log.info("已经生成目标表：" + table.getTargetTableName());
            //5、创建任务调度
            DataSynrecordEntity dataSynrecord = new DataSynrecordEntity();
            dataSynrecord.setId(IdWorker.getId());
            dataSynrecord.setTableId(table.getTableId());
            dataSynrecord.setCron(table.getCron());
            dataSynrecord.setDtId(ddsid);
            dataSynrecord.setDirId(prarmDto.getDtId());
            dataSynrecord.setSynType(prarmDto.getDataType());
            dataSynrecord.setSynWhere(table.getWhereColumn());
            dataSynrecord.setTargetTableName(table.getTargetTableName());
            String dataStr = JSON.toJSONString(dataSynrecord);
            //如果cron不为空创建任务
            if (StringUtils.isNotEmpty(table.getCron())) {
                XxlJobInfoEntity xxlJobInfoEntity = new XxlJobInfoEntity();
                xxlJobInfoEntity.setScheduleConf(table.getCron());
                xxlJobInfoEntity.setExecutorRouteStrategy("FIRST");
                xxlJobInfoEntity.setExecutorFailRetryCount(0);
                //xxlJobInfoEntity.setJobGroup(4);
                xxlJobInfoEntity.setJobGroup(groupEntity.getId());
                xxlJobInfoEntity.setMisfireStrategy("DO_NOTHING");
                xxlJobInfoEntity.setGlueType("BEAN");
                xxlJobInfoEntity.setScheduleType("CRON");
                xxlJobInfoEntity.setAlarmEmail("waytale@waytale.cn");
                xxlJobInfoEntity.setJobDesc("数据同步表:" + table.getTableName());
                xxlJobInfoEntity.setExecutorBlockStrategy("SERIAL_EXECUTION");
                xxlJobInfoEntity.setExecutorTimeout(0);
                xxlJobInfoEntity.setAuthor("waytale");
                xxlJobInfoEntity.setAddTime(new Date());
                xxlJobInfoEntity.setExecutorHandler("DataSynHandler");
                xxlJobInfoEntity.setExecutorParam(dataStr);

                Boolean aBoolean = xxlJobInfoService.saveOrUpdateXxlJob(xxlJobInfoEntity);
                log.info("任务调度id：" + xxlJobInfoEntity.getId());
                taskIds.add(xxlJobInfoEntity.getId());
                //6、回填tb_data_synrecord
                dataSynrecord.setTaskId(xxlJobInfoEntity.getId());
            }
            this.save(dataSynrecord);
            JSONObject jsonObject = JSONObject.parseObject(dataStr);
            //立即执行数据同步该方法已经处理顺序问题
            //kafkaTemplate.send(KafkaConstant.DATA_SYN_ACCEPT,jsonObject.toJSONString());
            //apiService.startDataSyn(jsonObject);
            apiService.execDataSyn(jsonObject, null);
        });
        //启动任务调度
        apiService.execXxlJobStart(taskIds);
        //this.startDataSyn(objectList);
    }

    /**
     * 创建数据同步任务每次循环开启新的事务
     *
     * @param dtId
     * @param dataType
     * @param dataSynDto
     * @return
     */
    //@Transactional(rollbackFor = {Exception.class, ServiceException.class},propagation = Propagation.NESTED)
    public String createDataSynTask(Long dtId, Integer dataType, DataSynDto dataSynDto) {
        if (ObjectUtils.isEmpty(dataSynDto)) {
            return null;
        }
        XxlJobGroupEntity groupEntity = groupService.getOne(Wrappers.<XxlJobGroupEntity>lambdaQuery().eq(XxlJobGroupEntity::getAppName, jobName).last("limit 1"));

        //1、回填tb_direct_data_set
        DirectDataSetEntity directDataSet = new DirectDataSetEntity();
        //生成id回填
        long ddsid = IdWorker.getId();
        directDataSet.setId(ddsid);
        directDataSet.setDtId(dtId);
        directDataSet.setTableName(dataSynDto.getTableName());
        directDataSet.setType("DB");
        TableEntity tableEntity = tableService.getById(dataSynDto.getTableId());
        directDataSet.setDsId(tableEntity.getDatasourceId());
        directDataSetService.save(directDataSet);
        //2、回填tb_directory
        DirectoryEntity directoryEntity = new DirectoryEntity();
        directoryEntity.setDtId(ddsid);
        directoryEntity.setPid(dtId);
        directoryEntity.setTableName(dataSynDto.getTargetTableName());
        directoryEntity.setIsDir(1);
        //设置名称先判断数据集是否存在
        long count = directoryService.count(Wrappers.<DirectoryEntity>lambdaQuery()
                .eq(DirectoryEntity::getPid, dtId)
                .eq(DirectoryEntity::getName, dataSynDto.getTableName())
        );
        //如果存在
        if (count > 0) {
            directoryEntity.setName(dataSynDto.getTableName() + "1");
        } else {
            directoryEntity.setName(dataSynDto.getTableName());
        }
        directoryService.save(directoryEntity);
        //3、回填tb_data_set_column
        try {
            List<DataSetColumnEntity> column = database.column(directDataSet);
            log.info("获取表字段信息：{}", column.toArray());
            dataSetColumnService.saveBatch(column);
        } catch (Exception e) {
            directDataSetService.removeById(ddsid);
            directoryService.remove(Wrappers.<DirectoryEntity>lambdaQuery().eq(DirectoryEntity::getDtId,ddsid));
            return dataSynDto.getTableName()+":获取源数据失败";
        }
        //4、先创建目标表
        //System.out.println(table.getTargetTableName());
        //jdbcService.createTableSourceDS(table.getTargetTableName(), table.getTabelId(), table.getWhereColumn());
        Boolean tableSourceDS = jdbcService.createTableSourceDS(dataSynDto.getTargetTableName(), ddsid, dataSynDto.getWhereColumn());
        log.info(dataSynDto.getTableName() + "创建目标表结果：" + tableSourceDS);
        if (!tableSourceDS) {
            directDataSetService.removeById(ddsid);
            directoryService.remove(Wrappers.<DirectoryEntity>lambdaQuery().eq(DirectoryEntity::getDtId,ddsid));
            return dataSynDto.getTableName()+":创建目标表失败";
        }

        log.info("已经生成目标表：" + dataSynDto.getTargetTableName());
        //5、创建任务调度
        DataSynrecordEntity dataSynrecord = new DataSynrecordEntity();
        dataSynrecord.setId(IdWorker.getId());
        dataSynrecord.setTableId(dataSynDto.getTableId());
        dataSynrecord.setCron(dataSynDto.getCron());
        dataSynrecord.setDtId(ddsid);
        dataSynrecord.setDirId(dtId);
        dataSynrecord.setSynType(dataType);
        dataSynrecord.setSynWhere(dataSynDto.getWhereColumn());
        dataSynrecord.setTargetTableName(dataSynDto.getTargetTableName());
        String dataStr = JSON.toJSONString(dataSynrecord);
        //如果cron不为空创建任务
        if (StringUtils.isNotEmpty(dataSynDto.getCron())) {
            XxlJobInfoEntity xxlJobInfoEntity = new XxlJobInfoEntity();
            xxlJobInfoEntity.setScheduleConf(dataSynDto.getCron());
            xxlJobInfoEntity.setExecutorRouteStrategy("FIRST");
            xxlJobInfoEntity.setExecutorFailRetryCount(0);
            //xxlJobInfoEntity.setJobGroup(4);
            xxlJobInfoEntity.setJobGroup(groupEntity.getId());
            xxlJobInfoEntity.setMisfireStrategy("DO_NOTHING");
            xxlJobInfoEntity.setGlueType("BEAN");
            xxlJobInfoEntity.setScheduleType("CRON");
            xxlJobInfoEntity.setAlarmEmail("waytale@waytale.cn");
            xxlJobInfoEntity.setJobDesc("数据同步表:" + dataSynDto.getTableName());
            xxlJobInfoEntity.setExecutorBlockStrategy("SERIAL_EXECUTION");
            xxlJobInfoEntity.setExecutorTimeout(0);
            xxlJobInfoEntity.setAuthor("waytale");
            xxlJobInfoEntity.setAddTime(new Date());
            xxlJobInfoEntity.setExecutorHandler("DataSynHandler");
            xxlJobInfoEntity.setExecutorParam(dataStr);

            Boolean aBoolean = xxlJobInfoService.saveOrUpdateXxlJob(xxlJobInfoEntity);
            log.info("任务调度id：" + xxlJobInfoEntity.getId());
            //taskIds.add(xxlJobInfoEntity.getId());
            //启动定时任务
            apiService.execXxlJobStartOne(xxlJobInfoEntity.getId());
            //6、回填tb_data_synrecord
            dataSynrecord.setTaskId(xxlJobInfoEntity.getId());
        }
        this.save(dataSynrecord);
        JSONObject jsonObject = JSONObject.parseObject(dataStr);
        //立即执行数据同步该方法已经处理顺序问题
        //kafkaTemplate.send(KafkaConstant.DATA_SYN_ACCEPT,jsonObject.toJSONString());
        //apiService.startDataSyn(jsonObject);
        apiService.execDataSyn(jsonObject, null);

        return dataSynDto.getTableName() + ":创建成功";
    }


    /**
     * 单表数据同步
     *
     * @param dirId
     * @param tableId
     */
    public void execOneTableSyn(Long dirId, Long tableId) {
        DataSynrecordEntity synrecordEntity = getOne(Wrappers.<DataSynrecordEntity>lambdaQuery().eq(DataSynrecordEntity::getTableId, tableId)
                .eq(DataSynrecordEntity::getDirId, dirId));
        if (ObjectUtil.isEmpty(synrecordEntity)) {
            throw new ServiceException("任务不存在");
        }
        String jsonString = JSON.toJSONString(synrecordEntity);
        //执行数据同步
        JSONObject jsonObject = JSONObject.parseObject(jsonString);
        apiService.execDataSyn(jsonObject, null);
    }

    /**
     * 修改调度
     *
     * @param id
     * @param cron
     * @return
     */
    @Transactional(rollbackFor = Exception.class)
    public boolean editDataSynrecord(Long id, String cron) {
        DataSynrecordEntity synrecordEntity = this.getById(id);
        if (ObjectUtil.isEmpty(synrecordEntity)) {
            return false;
        }

        XxlJobGroupEntity groupEntity = groupService.getOne(Wrappers.<XxlJobGroupEntity>lambdaQuery().eq(XxlJobGroupEntity::getAppName, jobName).last("limit 1"));
        //判断调度是否存在
        if (StringUtils.isNotEmpty(synrecordEntity.getCron())) {
            XxlJobInfoEntity xxlJobInfoEntity = xxlJobInfoService.getById(synrecordEntity.getTaskId());
            xxlJobInfoEntity.setScheduleConf(cron);
            xxlJobInfoService.saveOrUpdateXxlJob(xxlJobInfoEntity);
        } else {
            //创建调度
            String jsonString = JSON.toJSONString(synrecordEntity);
            TableEntity table = tableService.getById(synrecordEntity.getTableId());
            XxlJobInfoEntity xxlJobInfoEntity = new XxlJobInfoEntity();
            xxlJobInfoEntity.setScheduleConf(cron);
            xxlJobInfoEntity.setExecutorRouteStrategy("FIRST");
            xxlJobInfoEntity.setExecutorFailRetryCount(0);
            xxlJobInfoEntity.setJobGroup(groupEntity.getId());
            xxlJobInfoEntity.setMisfireStrategy("DO_NOTHING");
            xxlJobInfoEntity.setGlueType("BEAN");
            xxlJobInfoEntity.setScheduleType("CRON");
            xxlJobInfoEntity.setAlarmEmail("waytale@waytale.cn");
            xxlJobInfoEntity.setJobDesc("数据同步表:" + table.getTableName());
            xxlJobInfoEntity.setExecutorBlockStrategy("SERIAL_EXECUTION");
            xxlJobInfoEntity.setExecutorTimeout(0);
            xxlJobInfoEntity.setAuthor("waytale");
            xxlJobInfoEntity.setAddTime(new Date());
            xxlJobInfoEntity.setExecutorHandler("DataSynHandler");
            xxlJobInfoEntity.setExecutorParam(jsonString);
            //保存任务调度
            xxlJobInfoService.saveOrUpdateXxlJob(xxlJobInfoEntity);
            //启动任务调度
            xxlJobInfoService.start(xxlJobInfoEntity.getId());
        }

        synrecordEntity.setCron(cron);

        save(synrecordEntity);

        return true;
    }

    public void removeByDtId(Long dtId) {
        LambdaQueryWrapper<DataSynrecordEntity> qw = Wrappers.lambdaQuery();
        qw.eq(DataSynrecordEntity::getDtId, dtId);
        DataSynrecordEntity one = getOne(qw);
        if (null != one) {
            removeById(one.getId());
            if (null != one.getTaskId()) {
                try {
                    xxlJobInfoService.stop(one.getTaskId());
                    xxlJobInfoService.removeJob(one.getTaskId());
                } catch (Exception e) {
                }
            }
        }
    }
}




